package com.gjy.kafka.k10;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-03 10:55:03
 */
public class KafkaConfig {

    // Kafka服务器地址
    private static final String BOOTSTRAP_SERVERS = "192.168.253.136:9092";

    /**
     * 获取生产者配置
     */
    public static Properties getProducerConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可靠性配置
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // 等待所有副本确认
        props.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); // 保证顺序
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性

        // 性能优化
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 批量大小16KB
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10); // 等待时间10ms
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 缓冲区32MB

        // 压缩配置
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        return props;
    }

    /**
     * 获取消费者配置
     */
    public static Properties getConsumerConfig(String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return props;
    }

    /**
     * 获取消费者配置（自动提交）
     */
    public static Properties getAutoCommitConsumerConfig(String groupId) {
        Properties props = getConsumerConfig(groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        return props;
    }

    /**
     * 获取消费者配置（手动提交）
     */
    public static Properties getManualCommitConsumerConfig(String groupId) {
        Properties props = getConsumerConfig(groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 从最新开始消费

        // 会话和心跳配置
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");

        // 一次poll的最大记录数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

        return props;
    }

    /**
     * 获取高性能消费者配置
     */
    public static Properties getHighPerformanceConsumerConfig(String groupId) {
        Properties props = getManualCommitConsumerConfig(groupId);

        // 优化配置
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 最小抓取字节
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 100); // 最大等待时间
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576); // 每个分区抓取大小

        return props;
    }

}
